package org.luxor.common.netty.protocol.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.DecoderException;
import io.netty.handler.codec.ReplayingDecoder;
import io.netty.handler.codec.TooLongFrameException;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.ObjectUtil;
import org.luxor.common.netty.constant.NettyReasonCode;
import org.luxor.common.netty.protocol.*;
import org.luxor.common.netty.protocol.message.NettyMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * NettyMessage 解码器
 *
 * @author Mr.yan @date 2024/1/7
 * @see io.netty.handler.codec.mqtt.MqttDecoder 参考Mqtt消息解码器
 * 参考 http://docs.oasis-open.org/mqtt/mqtt/v5.0/csprd02/mqtt-v5.0-csprd02.html#_Toc498345306
 */
public class NettyDecoder extends ReplayingDecoder<NettyDecoder.DecoderState> {
    private static final Logger log = LoggerFactory.getLogger(NettyDecoder.class);

    /**
     * 解码器的状态.我们从READ_FIXED_HEADER开始,然后是READ_VARIABLE_HEADER,最后是READ_PAYLOAD.
     */
    enum DecoderState {
        READ_FIXED_HEADER,
        READ_VARIABLE_HEADER,
        READ_PAYLOAD,
        BAD_MESSAGE,
    }

    private NettyFixedHeader nettyFixedHeader;
    private Object variableHeader;
    private int bytesRemainingInVariablePart;

    private final long packetSizeMax;

    public NettyDecoder(long packetSizeMax) {
        super(DecoderState.READ_FIXED_HEADER);
        this.packetSizeMax = ObjectUtil.checkPositive(packetSizeMax, "packetSizeMax");
    }

    @Override
    public void decode(ChannelHandlerContext ctx, ByteBuf buffer, List out) throws Exception {
        switch (state()) {
            case READ_FIXED_HEADER:
                try {
                    nettyFixedHeader = decodeFixedHeader(buffer);
                    bytesRemainingInVariablePart = nettyFixedHeader.remainingLength();
                    checkpoint(NettyDecoder.DecoderState.READ_VARIABLE_HEADER);
                    // fall through
                } catch (Exception cause) {
                    out.add(invalidMessage(ctx, cause));
                    return;
                }

            case READ_VARIABLE_HEADER:
                try {
                    final Result<?> decodedVariableHeader = decodeVariableHeader(buffer, nettyFixedHeader);
                    variableHeader = decodedVariableHeader.value;
                    if (bytesRemainingInVariablePart > packetSizeMax) {
                        buffer.skipBytes(actualReadableBytes());
                        throw new TooLongFrameException("too large message: " + bytesRemainingInVariablePart + " bytes");
                    }
                    bytesRemainingInVariablePart -= decodedVariableHeader.numberOfBytesConsumed;
                    checkpoint(NettyDecoder.DecoderState.READ_PAYLOAD);
                    // fall through
                } catch (Exception cause) {
                    out.add(invalidMessage(ctx, cause));
                    return;
                }

            case READ_PAYLOAD:
                try {
                    final Result<ByteBuf> decodedPayload = decodePayload(buffer, nettyFixedHeader.messageType(), bytesRemainingInVariablePart);
                    bytesRemainingInVariablePart -= decodedPayload.numberOfBytesConsumed;
                    if (bytesRemainingInVariablePart != 0) {
                        throw new DecoderException(
                                "non-zero remaining payload bytes: " +
                                        bytesRemainingInVariablePart + " (" + nettyFixedHeader.messageType() + ')');
                    }
                    checkpoint(NettyDecoder.DecoderState.READ_FIXED_HEADER);
                    NettyMessage message = NettyMessageFactory.newMessage(nettyFixedHeader, variableHeader, decodedPayload.value);
                    nettyFixedHeader = null;
                    variableHeader = null;
                    out.add(message);
                    break;
                } catch (Exception cause) {
                    out.add(invalidMessage(ctx, cause));
                    return;
                }

            case BAD_MESSAGE:
                // Keep discarding until disconnection.
                buffer.skipBytes(actualReadableBytes());
                break;

            default:
                // Shouldn't reach here.
                throw new Error();
        }
    }

    private NettyMessage invalidMessage(ChannelHandlerContext ctx, Exception cause) {
        log.error("因出现'无效报文',导致(NettyDecoder)解码异常.", cause);
        checkpoint(DecoderState.BAD_MESSAGE);
        NettyReasonCodeVariableHeader reasonVariableHeader = NettyReasonCodeVariableHeader.from(NettyReasonCode.INVALID_MESSAGE);
        return new NettyMessage(nettyFixedHeader, reasonVariableHeader, null);
    }

    private NettyFixedHeader decodeFixedHeader(ByteBuf buffer) {
        short b1 = buffer.readUnsignedByte();

        NettyMessageType messageType = NettyMessageType.valueOf(b1 >> 4);
        boolean dupFlag = (b1 & 0x08) == 0x08;
        int qosLevel = (b1 & 0x06) >> 1;

        int remainingLength = 0;
        int multiplier = 1;
        short digit;
        int loops = 0;
        do {
            digit = buffer.readUnsignedByte();
            remainingLength += (digit & 127) * multiplier;
            multiplier *= 128;
            loops++;
        } while ((digit & 128) != 0 && loops < 4);

        // Netty protocol limits Remaining Length to 4 bytes
        if (loops == 4 && (digit & 128) != 0) {
            throw new DecoderException("remaining length exceeds 4 digits (" + messageType + ')');
        }
        return new NettyFixedHeader(messageType, dupFlag, NettyQoS.valueOf(qosLevel), remainingLength);
    }

    private Result<?> decodeVariableHeader(ByteBuf buffer, NettyFixedHeader nettyFixedHeader) {
        switch (nettyFixedHeader.messageType()) {
            case CONNECT:
                return decodeConnectionVariableHeader(buffer);
            case CONN_ACK:
            case DISCONNECT:
                return decodeReasonCodeVariableHeader(buffer);
            case PUBLISH:
                return decodeSendVariableHeader(buffer);
            case PUB_REL:
                return decodeSingleMessageIdVariableHeader(buffer);
            case PUB_ACK:
            case PUB_REC:
            case PUB_COMP:
                return decodeSendReplyVariableHeader(buffer);
            case PING:
            case PONG:
                // Empty variable header
                return new Result<>(null, 0);
            default:
                //shouldn't reach here
                throw new DecoderException("Unknown message type: " + nettyFixedHeader.messageType());
        }
    }

    private static Result<NettyConnectVariableHeader> decodeConnectionVariableHeader(ByteBuf buffer) {
        final NettyDecoder.Result<String> clientString = decodeString(buffer);
        final String clientId = clientString.value;
        int numberOfBytesConsumed = clientString.numberOfBytesConsumed;

        final NettyDecoder.Result<String> usernameString = decodeString(buffer);
        final String username = usernameString.value;
        numberOfBytesConsumed += usernameString.numberOfBytesConsumed;

        final NettyDecoder.Result<String> passwordString = decodeString(buffer);
        final String password = passwordString.value;
        numberOfBytesConsumed += passwordString.numberOfBytesConsumed;

        final NettyConnectVariableHeader nettyConnectVariableHeader = NettyConnectVariableHeader.from(clientId, username, password);
        return new Result<>(nettyConnectVariableHeader, numberOfBytesConsumed);
    }

    private Result<NettyReasonCodeVariableHeader> decodeReasonCodeVariableHeader(ByteBuf buffer) {
        int errorCode = buffer.readInt();
        int numberOfBytesConsumed = 4;

        final NettyDecoder.Result<String> errorMessageString = decodeString(buffer);
        final String errorMessage = errorMessageString.value;
        numberOfBytesConsumed += errorMessageString.numberOfBytesConsumed;

        final NettyReasonCodeVariableHeader nettyReasonCodeVariableHeader = NettyReasonCodeVariableHeader.from(errorCode, errorMessage);
        return new Result<>(nettyReasonCodeVariableHeader, numberOfBytesConsumed);
    }

    private Result<NettySendVariableHeader> decodeSendVariableHeader(ByteBuf buffer) {
        final NettyDecoder.Result<String> messageIdString = decodeString(buffer);
        final String messageId = messageIdString.value;
        int numberOfBytesConsumed = messageIdString.numberOfBytesConsumed;

        final NettyDecoder.Result<String> messageNameString = decodeString(buffer);
        final String messageName = messageNameString.value;
        numberOfBytesConsumed += messageNameString.numberOfBytesConsumed;

        final NettyDecoder.Result<String> sessionIdString = decodeString(buffer);
        final String sessionId = sessionIdString.value;
        numberOfBytesConsumed += sessionIdString.numberOfBytesConsumed;

        final NettySendVariableHeader sendVariableHeader = NettySendVariableHeader.from(messageId, messageName, sessionId);
        return new Result<>(sendVariableHeader, numberOfBytesConsumed);
    }

    private Result<NettyMessageIdVariableHeader> decodeSingleMessageIdVariableHeader(ByteBuf buffer) {
        final NettyDecoder.Result<String> messageIdString = decodeString(buffer);
        final String messageId = messageIdString.value;
        int numberOfBytesConsumed = messageIdString.numberOfBytesConsumed;

        final NettyMessageIdVariableHeader messageIdVariableHeader = NettyMessageIdVariableHeader.from(messageId);
        return new Result<>(messageIdVariableHeader, numberOfBytesConsumed);
    }

    private Result<NettySendReplyMessageVariableHeader> decodeSendReplyVariableHeader(ByteBuf buffer) {
        final NettyDecoder.Result<String> messageIdString = decodeString(buffer);
        final String messageId = messageIdString.value;
        int numberOfBytesConsumed = messageIdString.numberOfBytesConsumed;

        int errorCode = buffer.readInt();
        numberOfBytesConsumed += 4;

        final NettyDecoder.Result<String> errorMessageString = decodeString(buffer);
        final String errorMessage = errorMessageString.value;
        numberOfBytesConsumed += errorMessageString.numberOfBytesConsumed;

        final NettySendReplyMessageVariableHeader sendReplyMessageVariableHeader = NettySendReplyMessageVariableHeader.from(messageId, errorCode, errorMessage);
        return new Result<>(sendReplyMessageVariableHeader, numberOfBytesConsumed);
    }

    private Result<ByteBuf> decodePayload(ByteBuf buffer, NettyMessageType messageType, int bytesRemainingInVariablePart) {
        switch (messageType) {
            case PUBLISH:
            case PUB_ACK:
            case PUB_REC:
            case PUB_COMP:
                return decodeSendPayload(buffer, bytesRemainingInVariablePart);
            default:
                // unknown payload , no byte consumed
                return new Result<>(null, 0);
        }
    }


    private static Result<ByteBuf> decodeSendPayload(ByteBuf buffer, int bytesRemainingInVariablePart) {
        ByteBuf b = buffer.readRetainedSlice(bytesRemainingInVariablePart);
        return new Result<>(b, bytesRemainingInVariablePart);
    }


    private static Result<String> decodeString(ByteBuf buffer) {
        return decodeString(buffer, 0, Integer.MAX_VALUE);
    }

    private static Result<String> decodeString(ByteBuf buffer, int minBytes, int maxBytes) {
        int size = decodeMsbLsb(buffer);
        int numberOfBytesConsumed = 2;
        if (size < minBytes || size > maxBytes) {
            buffer.skipBytes(size);
            numberOfBytesConsumed += size;
            return new Result<>(null, numberOfBytesConsumed);
        }
        String s = buffer.toString(buffer.readerIndex(), size, CharsetUtil.UTF_8);
        buffer.skipBytes(size);
        numberOfBytesConsumed += size;
        return new Result<>(s, numberOfBytesConsumed);
    }

    /**
     * numberOfBytesConsumed = 2. return decoded result.
     */
    private static int decodeMsbLsb(ByteBuf buffer) {
        int min = 0;
        int max = 65535;
        short msbSize = buffer.readUnsignedByte();
        short lsbSize = buffer.readUnsignedByte();
        int result = msbSize << 8 | lsbSize;
        if (result < min || result > max) {
            result = -1;
        }
        return result;
    }


    private static final class Result<T> {

        private final T value;
        private final int numberOfBytesConsumed;

        Result(T value, int numberOfBytesConsumed) {
            this.value = value;
            this.numberOfBytesConsumed = numberOfBytesConsumed;
        }
    }
}
